package cc.verywell.pureblock.net;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.net.Timelog.BasicTimelog;
import cc.verywell.pureblock.net.Timelog.TimelogGenerator;
import cc.verywell.pureblock.util.EncryptUtil;

/**
* 
* Copyright 2019 TanYaqiu
* @author TanYaqiu 
* @address flat 601
* @date 2019年3月15日 上午1:21:29
*/
public class AioProvider  implements Closeable,ConnectionFreeProvider{
	public final ThreadGroup THREADGROUP = new ThreadGroup("PureBlock");

	public static InetSocketAddress BIND_PORT = new InetSocketAddress(10090) ;// 绑定端口号， 
	public static int MAX_CONNECTIONS = 500 ;// 500==DEFAULT; eth默认100 ;最大连接数; 
	public static int MAX_PENDING_CONNECTIONS = 50 ;// 50==DEFAULT; eth默认30 ;连接等待数序列，设置这个可以预防攻击?真遇到攻击，请用负载均衡去挡;
	public static int MAX_HANDSHAKE_CHANNEL_CACHE = MAX_PENDING_CONNECTIONS/10;// DEFAULT==MAX_PENDING_CONNECTIONS
	public static int FILE_MAX_SIZE = 1024*512 ;// 512K+64btyes==DEFAULT; 文件的最大大小， 
	public static int MESSAGE_SIZE = FILE_MAX_SIZE+64 ;// 512K+64btyes==DEFAULT; 收到包的最大大小， 
	public static int TIME_OUT = 10 ;// 超时timeOut，以秒为单位, 更新后，请重启AioServer
	public static int CLOSE_TIME_OUT = 120 ;// 关闭前的发送超时，以秒为单位
	public static long RECONNECT_TIME_OUT = 30*1000 ;// 主动连接间隔，以豪秒为单位
	@Override
	public boolean isTooManyPendingConnections() {
		int pendingSize = pendingSet.size();
		if (pendingSize >= AioProvider.MAX_PENDING_CONNECTIONS) {
			if (Pblog.IO)
				Pblog.IO("TOO MANY PENDING CONNECTIONS");
			return true;
		}
		if (connectSet.size() >= AioProvider.MAX_CONNECTIONS - pendingSize) {
			if (Pblog.IO)
				Pblog.IO("TOO MANY CONNECTIONS");
			return true;
		}
		return false;
	}
	
	private AsynchronousChannelGroup acg;
	private ExecutorService es_of_acg;
	public synchronized AsynchronousChannelGroup getDefaultChannelGroup(int howManyThread) throws IOException{
		if(acg==null) {
			acg = AsynchronousChannelGroup.withThreadPool(getExecutor(howManyThread));
			//请不要使用这个方法(因不能释放线程)：AsynchronousChannelGroup.withFixedThreadPool(nThreads, threadFactory);
		}
		return acg;
	}
	public synchronized ExecutorService getExecutor(int howManyThread) {
		if(es_of_acg==null) {
			final int threads = howManyThread>0?howManyThread:Runtime.getRuntime().availableProcessors();
			es_of_acg = Executors.newFixedThreadPool(threads, new ThreadFactory() {
				private AtomicInteger count = new AtomicInteger();
				@Override
				public Thread newThread(Runnable r) {
					Thread t = new Thread(THREADGROUP,r, "fixedAioServer_"+count.getAndIncrement()+"of"+threads);
					t.setPriority(t.getPriority()-1); //IO操作要慢于业务需要，
					return t;
				}
			});
		}
		return es_of_acg;
	}
	
	final private Context context;
	final private Object myAlias;
	Set<NormalChannel> connectSet;
	Set<AsynchronousSocketChannel> pendingSet;
	public AioProvider(Context context,Object myAlias) {
		this.context = context;
		this.myAlias = myAlias;
		pendingSet = Collections.synchronizedSet(new HashSet<AsynchronousSocketChannel>(AioProvider.MAX_PENDING_CONNECTIONS));
		connectSet = Collections.synchronizedSet(new HashSet<NormalChannel>());
		this.connectionFreeListener = new HashSet<>(0);
	}
	public Set<NormalChannel> getConnectSet(){
		return this.connectSet;
	}
	public Set<AsynchronousSocketChannel> getPendingSet(){
		return this.pendingSet;
	}
	public Object myAlias() {
		return this.myAlias;
	}

	private BlockingQueue<NormalChannel> normalChannels = new LinkedBlockingQueue<>();
	/**
	 * @param channel
	 * @param eu
	 * @param unread
	 * @return 如果newSession出错会返回null, 否则返回正常session;
	 */
	public NormalChannel handshakeCompleted(AsynchronousSocketChannel channel, EncryptUtil eu, ByteBuffer unread, NormalFunction closeDo, int salt,boolean isClient) {
		if(closeDo==null)closeDo=closeWhatWeDo;else closeDo.addNext(closeWhatWeDo);
		//try {
			AsynchronousSocketChannel removechannel = channel;
			NormalChannel nc = normalChannels.poll();
			if(nc==null)nc = new NormalChannel(myAlias());
			if(unread!=null)unread.mark();
			try {
				NormalChannel removenc = null;
				try {
					connectSet.add(removenc = nc); //必须要在init之前记入，因为newSession可能会发生在init之中;
					pendingSet.remove(removechannel);
					removechannel=null;
					nc.init(channel, context, eu, unread, closeDo, salt, isClient); //channel没清干净?
					removenc = null;
					return nc;
				} catch (IOException e) {
					nc.clearUser();
					if(Pblog.IO)Pblog.IO(e);
					if(unread!=null)unread.reset();
					nc = new NormalChannel(myAlias());
					connectSet.add(nc); //必须要在init之前记入，因为newSession可能会发生在init之中;
					connectSet.remove(removenc);
					removenc = nc;
					nc.init(channel, context, eu, unread, closeDo, salt, isClient); //newSession会出错, 然后由下边关闭channel;
					removenc = null;
					return nc;
				} finally {
					if(removenc!=null)connectSet.remove(removenc);
				}
			} catch (Throwable e1) {
				if(Pblog.IO)Pblog.IO(e1); //致命错误注意;
				System.err.println("structure bug happen time "+System.currentTimeMillis());
				e1.printStackTrace();
				try {
					nc.close(true); //此时channel可能为NULL;
				}catch(Throwable e2) {if(Pblog.DEBUG)Pblog.DEBUG(e2);}
				return null;
			} finally {
				if(removechannel!=null)pendingSet.remove(removechannel);
			}
			
		//	if(nc!=null)connectSet.add(nc);
		//	return nc;
		//}finally { //常见于newSession出错;
		//	if(null!=channel)pendingSet.remove(channel);
		//}
	}
	public void handshakeFaild(AsynchronousSocketChannel channel, Throwable exp) {
		pendingSet.remove(channel);
		throwConnectionFree();///或主动连接下一个;
	}
	
	private NormalFunction closeWhatWeDo = new NormalFunction() {
		@Override
		public void addNext(NormalFunction function) {
			throw new IllegalAccessError();
		}
		@Override
		public void init(NormalSession channel) {}
		@Override
		public void closed(NormalChannel channel) {
			connectSet.remove(channel);
			normalChannels.offer(channel.clearUser());
			throwConnectionFree();///或主动连接下一个;
		}
	};

	
	/**
	 * 连接空位提示;
	 */
	@Override
	public void addConnectionFreeListener(ConnectionFreeListener l) {
		synchronized (connectionFreeListener) {
			connectionFreeListener.add(l);
		}
	}
	@Override
	public void removeConnectionFreeListener(ConnectionFreeListener l) {
		synchronized (connectionFreeListener) {
			connectionFreeListener.remove(l);
		}
	}
	final private Collection<ConnectionFreeListener> connectionFreeListener;
	private void throwConnectionFree() {
		if(isTooManyPendingConnections())return;
		synchronized (connectionFreeListener) {
			for (ConnectionFreeListener l : connectionFreeListener) {
				l.connectionfree();
			}
		}
	}
	


	
	@Override
	public void close() {
		for (NormalChannel normalChannel : new ArrayList<>(connectSet)) {
			normalChannel.close();
		}
		for (AsynchronousSocketChannel channel : new ArrayList<>(pendingSet)) {
			try {
				channel.shutdownInput();
			} catch (IOException e) {
				if(Pblog.IO)Pblog.IO("clos@",this,": ",e);
			}
			try {
				channel.shutdownOutput();
			} catch (IOException e) {
				if(Pblog.IO)Pblog.IO("clos@",this,": ",e);
			}
			try {
				channel.close();
			} catch (IOException e) {
				if(Pblog.IO)Pblog.IO("clos@",this,": ",e);
			}
		}
		if(acg!=null)try{
			acg.shutdown();
			try {
				acg.awaitTermination(10, TimeUnit.SECONDS);
			} catch (InterruptedException e) {
				if(Pblog.IO)Pblog.IO(e);
			}
		}finally {acg=null;}
		//实测acg不会自动关闭线程;
		if(es_of_acg!=null)try{
			es_of_acg.shutdown();
			try {
				es_of_acg.awaitTermination(10, TimeUnit.SECONDS);
			} catch (InterruptedException e) {
				if(Pblog.IO)Pblog.IO(e);
			}
		}finally {es_of_acg=null;}

	}
	
	/**
	 * 默认的方法， 就是生成一个BasicTimelog;
	 */
	static public TimelogGenerator timelog = new TimelogGenerator(){
		@Override
		public Timelog getTimelog(NormalSession nss) {
			return new BasicTimelog();
		}
	};

}
